package com.lagou.hbase.hw;



import org.apache.hadoop.hbase.Cell;
import org.apache.hadoop.hbase.CellUtil;
import org.apache.hadoop.hbase.TableName;
import org.apache.hadoop.hbase.client.*;
import org.apache.hadoop.hbase.coprocessor.BaseRegionObserver;
import org.apache.hadoop.hbase.coprocessor.ObserverContext;
import org.apache.hadoop.hbase.coprocessor.RegionCoprocessorEnvironment;
import org.apache.hadoop.hbase.regionserver.wal.WALEdit;
import org.apache.hadoop.hbase.util.Bytes;


import org.mortbay.log.Log;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

import java.io.IOException;
import java.util.*;


public class MyFriendProcessor extends BaseRegionObserver {


    private static final Logger LOG = LoggerFactory.getLogger(MyFriendProcessor.class);

    private static final String MYF="myf";

    @Override
    public void preDelete(ObserverContext<RegionCoprocessorEnvironment> e, Delete delete, WALEdit edit, Durability durability) throws IOException {
        //当f1 删除一条数据的 rk : u1 myf：u2 u2
//        // 同时 f1 也要删除一条数据 rk u2 myf：u1 u1

        HTableInterface y1 = e.getEnvironment().getTable(TableName.valueOf("y1"));


        try {
            //看看哪个好友删除你
            byte[] row = delete.getRow();
            LOG.info("your f ：" + new String(row));
            ResultScanner myf = y1.getScanner(Bytes.toBytes("myf"));
            List<String> rks = new ArrayList<String>();
            for (Result result : myf) {
                String rk = new String(result.getRow());
                Log.info("rk=====>{}",rk);
                rks.add(rk);
            }
            Log.info("rk size  ====》" + rks.size());

            List<Cell> cells = delete.getFamilyCellMap().get(MYF.getBytes());
            if (cells == null || cells.size() == 0) {
                LOG.info("NO myf famally operation ,just do it.");
                return;
            }
            Log.info("cells size===>{}", cells.size());

            LOG.info("myf hava operation ");
            // 开始检查好朋友内的操作情况
            byte[] qualifierName = null;
            boolean aDeleteFlg = false;//对方删除我，我也要删除对方
            for (Cell cell : cells) {
                qualifierName = CellUtil.cloneQualifier(cell);//y3

                LOG.info("myf hava operation " + new String(qualifierName));
                if (rks.contains(new String(qualifierName)) ) {
                    LOG.info("find myf del me====>" + new String(row));
                    aDeleteFlg = true;
                }
            }

            // 如果对于对方删除，则需要对我也要删除对方
            if (aDeleteFlg) {
                LOG.info("i also been deleted!");
                Delete delete1 = new Delete(qualifierName);
                delete.addColumn(Bytes.toBytes(MYF), row);
                y1.delete(delete1);
                LOG.info("deleted sccuee");
            }
        } catch (Exception ex) {
            LOG.error(ex.getMessage());
        }
    }
}
